import com.zuikaku.pojo.Order;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.math.BigDecimal;
import java.util.UUID;

public class KeyByDemo {
    public static void main(String[] args) {

        //创建环境
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.createLocalEnvironment();
        environment.setParallelism(1);

        //根据order获取source
        DataStreamSource<Order> orderDS = environment.fromElements(new Order(UUID.randomUUID().toString(),"HuaWei Mate60",1003,new BigDecimal("6999.99")),
                new Order(UUID.randomUUID().toString(),"iPhone 15 Pro Max",1002,new BigDecimal("6979.99")),
                new Order(UUID.randomUUID().toString(),"HuaWei Mate60",1001,new BigDecimal("8999.99")),
                new Order(UUID.randomUUID().toString(),"Nokia X1",1004,new BigDecimal("1999.99")),
                new Order(UUID.randomUUID().toString(),"Nokia X1",1005,new BigDecimal("1969.99")));

        //demo1:根据itemName进行分组统计金额
        //由于bigDecimal类型无法直接+-*/，因此先map转换为基本数据类型包装类的元组操作
        DataStream<Tuple2<String, Float>> itemPriceDS = orderDS.map(new MapFunction<Order, Tuple2<String, Float>>() {
            @Override
            public Tuple2<String, Float> map(Order order) throws Exception {
                return new Tuple2<>(order.getItemName(),order.getPrice().floatValue());
            }
        });

        //进行分组<数据的泛型，key的泛型>
        KeyedStream<Tuple2<String, Float>, String> keyDs = itemPriceDS.keyBy(new KeySelector<Tuple2<String, Float>, String>() {
            @Override
            public String getKey(Tuple2<String, Float> stringFloatTuple2) throws Exception {
                return stringFloatTuple2.f0;
            }
        });
        DataStream<Tuple2<String, Float>> sumDs = keyDs.sum(1);//对Tuple2中的float进行求和
        sumDs.print("分组求和后");

        //demo2:对itemName进行分组，拿到每种商品最新下单的用户id（用户id视为自增，最大为最新）
        KeyedStream<Order, String> keyByItemDs = orderDS.keyBy(new KeySelector<Order, String>() {
            @Override
            public String getKey(Order order) throws Exception {
                return order.getItemName();
            }
        });
        DataStream<Order> latestCustomerDs = keyByItemDs.max("customerId");
        latestCustomerDs.print("当前商品最新用户");

        try {
            environment.execute("key by demo");
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
